package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TradeOrderBean;
import com.zhu.config.KafkaTopicConfig;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.TimestampLtz3CompareUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 10.8 交易域下单各窗口汇总表
 *基于  dwd_trade_detail_app  基于 process  left join 存在  重复数据
 */
public class DWSTradeOrderWindowApp {

    public static void main(String[] args) throws Exception {
        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);
        //checkPoint
       /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次
        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
         */

        //todo get_kafka_dwd_trade_order_detail
        String topic = KafkaTopicConfig.KAFKA_DWD_TRADE_ORDER_DETAIL_TOPIC;
        String groupId = "dws_trade_order_window" + KafkaTopicConfig.KAFKA_GROUP_ID_LAST_NAME;
        DataStreamSource<String> kafkaStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo json
        SingleOutputStreamOperator<JSONObject> jsonObjDStream = kafkaStream.flatMap(
                new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String data, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(data);
                            collector.collect(jsonObject);
                        } catch (Exception exception) {
                            System.out.println("Dirty_Data>>>" + data);
                            exception.printStackTrace();
                        }
                    }
                }
        );

        //todo order_detail_id 分组
        KeyedStream<JSONObject, String> keyedByDetailIdDStream =
                jsonObjDStream.keyBy(jsonObject -> jsonObject.getString("id"));
        //keyedByDetailIdDStream.print(">>>>>");


        //todo 去重
        //选择最晚来的一条数据
        SingleOutputStreamOperator<JSONObject> filterOWNDetailDataDStream = keyedByDetailIdDStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

            //维护状态
            private ValueState<JSONObject> lastDetailData = null;


            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ValueStateDescriptor<JSONObject> orderDetailStateDescriptor =
                        new ValueStateDescriptor<>("last-join-data", JSONObject.class);
                lastDetailData = getRuntimeContext().getState(orderDetailStateDescriptor);
            }

            @Override
            public void processElement(JSONObject data, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String nowProcessTime = data.getString("row_op_ts");
                 JSONObject lastJsonData = lastDetailData.value();
                if ( lastJsonData == null) {
                    lastDetailData.update(data);
                    //timeService
                    context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime() + 5000L);
                } else {
                    String lastProcessTime = lastJsonData.getString("row_op_ts");
                    int castValue = TimestampLtz3CompareUtil.compare(lastProcessTime, nowProcessTime);
                    if (castValue != 1) {
                        lastDetailData.update(data);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                //定时器触发  写出结果 清空状态
                out.collect(lastDetailData.value());
                lastDetailData.clear();
            }
        });


        //todo waterMark
        SingleOutputStreamOperator<JSONObject> detailDataWithWaterMarkDStream = filterOWNDetailDataDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return DataFormatUtil.toTs(jsonObject.getString("create_time"),true);
                    }
                }));

        //todo 分组
        KeyedStream<JSONObject, String> keyedByUserDetailDataDStream
                = detailDataWithWaterMarkDStream.keyBy(jsonObject -> jsonObject.getString("user_id"));

        //todo 筛选独立下单  flatMap 也可以
        SingleOutputStreamOperator<TradeOrderBean> newOrUniqueUserDetailDStream = keyedByUserDetailDataDStream.flatMap(

                //维护用户独立订单状态 当日首日 data_id create_time 比较当日独立下单的用户和独支付的用户

                new RichFlatMapFunction<JSONObject, TradeOrderBean>() {

                    private ValueState<String> lastDetailState;


                    @Override
                    public void flatMap(JSONObject jsonObject, Collector<TradeOrderBean> collector) throws Exception {

                        String creatTime = jsonObject.getString("create_time").split(" ")[0];
                        String lastOrderDate = lastDetailState.value();
                        long orderNewUserCount = 0L;
                        long orderUniqueUserCount = 0L;
                        Double splitActivityAmount =
                                jsonObject.getDouble("split_activity_amount");
                        Double splitCouponAmount =
                                jsonObject.getDouble("split_coupon_amount");
                        Double splitOriginalAmount =
                                jsonObject.getDouble("split_original_amount");
                        // Long ts = jsonObject.getLong("ts") * 1000L

                        if (lastOrderDate == null) {
                            orderNewUserCount = 1L;
                            orderUniqueUserCount = 1L;
                            lastDetailState.update(creatTime);
                        } else {
                            if (!creatTime.equals(lastOrderDate)) {
                                orderUniqueUserCount = 1L;
                                lastDetailState.update(creatTime);
                            }
                        }
                        if (orderUniqueUserCount != 0L) {
                            collector.collect(new TradeOrderBean(
                                    null, null,
                                    orderUniqueUserCount, orderNewUserCount,
                                    (splitActivityAmount == null ? 0.0 : splitActivityAmount) ,
                                    (splitCouponAmount == null ? 0.0 : splitCouponAmount),
                                    (splitOriginalAmount == null ? 0.0 : splitOriginalAmount),
                                    null));
                        }
                    }

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastDetailState =
                                getRuntimeContext().getState(new ValueStateDescriptor<String>("user_last_detail_date", String.class));

                    }
                }
        );


        //todo window
        AllWindowedStream<TradeOrderBean, TimeWindow> windowNewOrUniqueUserDetailDStream
                = newOrUniqueUserDetailDStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));


        //todo reduce
        SingleOutputStreamOperator<TradeOrderBean> resultDStream = windowNewOrUniqueUserDetailDStream.reduce(
                new ReduceFunction<TradeOrderBean>() {
                    @Override
                    public TradeOrderBean reduce(TradeOrderBean value1, TradeOrderBean value2) throws Exception {
                        value1.setOrderNewUserCount(value1.getOrderNewUserCount() + value2.getOrderNewUserCount());
                        value1.setOrderUniqueUserCount(value2.getOrderUniqueUserCount() + value2.getOrderUniqueUserCount());
                        value1.setOrderActivityReduceAmount(value1.getOrderActivityReduceAmount() + value2.getOrderActivityReduceAmount());
                        value1.setOrderCouponReduceAmount(value1.getOrderCouponReduceAmount() + value2.getOrderCouponReduceAmount());
                        value1.setOrderOriginalTotalAmount(value1.getOrderOriginalTotalAmount() + value2.getOrderOriginalTotalAmount());
                        return value1;
                    }
                },
                new AllWindowFunction<TradeOrderBean, TradeOrderBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TradeOrderBean> iterable, Collector<TradeOrderBean> collector) throws Exception {
                        String start = DataFormatUtil.toYmdHms(timeWindow.getStart());
                        String end = DataFormatUtil.toYmdHms(timeWindow.getEnd());
                        long ts = System.currentTimeMillis();
                        TradeOrderBean next = iterable.iterator().next();
                        next.setTs(ts);
                        next.setEdt(end);
                        next.setStt(start);
                        collector.collect(next);
                    }
                }
        );

        //todo clickHouse
        resultDStream.print("dws_order_detail_window >>>");
        resultDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction(
                "insert into dws_trade_order_window  values(?,?,?,?,?,?,?,?)"));


        //todo job
        streamExecutionEnvironment.execute();
    }
}
